package mxx.mr.job;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

/**
 * @program:report-form-hadoop
 * @description:设备综合效率OEE报表Job
 * @author:AlTuring
 * @Date:2019/11/12-11:14
 **/
public class DeviceComplexEfficiencyOeeJob {
    /**
     * 设备综合效率OEE报表-Mapper
     * 1.设备OEE
     */
    public static class DeviceComplexEfficiencyOeeMapper extends Mapper<LongWritable,Text,Text, MapWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            MapWritable mw = new MapWritable();
            // -- 构造key
            String[] datas = value.toString().split(",");
            //设备SEQ
            String deviceSeq = datas[0];
            //时间（2019-10-01）
            String time = datas[7].substring(0,10);
            //自定义key
            Text outKey = new Text(deviceSeq +"_"+
                    time);
            // -- 写出
            mw.put(new IntWritable(1),value);
            context.write(outKey, mw);
        }
    }

    /**
     * 设备综合效率OEE报表-Mapper
     * 2.不合格工件数量
     */
    public static class DeviceComplexEfficiencyOeeUnqualifiedMapper extends Mapper<LongWritable,Text,Text, MapWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            MapWritable mw = new MapWritable();
            // -- 构造key
            String[] datas = value.toString().split(",");
            //设备SEQ
            String deviceSeq = datas[0];
            //检验时间（2019-10-01）
            String time = datas[5].substring(0,10);

            //定义key
            Text outKey = new Text(deviceSeq +"_"+
                    time);

            mw.put(new IntWritable(2),value);
            context.write(outKey, mw);

        }
    }

    /**
     * 设备综合效率OEE报表-Mapper
     * 3.总零件数
     */
    public static class DeviceComplexEfficiencyOeeTotalMapper extends Mapper<LongWritable,Text,Text, MapWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            MapWritable mw = new MapWritable();
            // -- 构造key
            String[] datas = value.toString().split(",");
            //设备SEQ
            String deviceSeq = datas[0];
            //加工完成的时间（2019-10-01）
            String time = datas[5].substring(0,10);

            //定义key
            Text outKey = new Text(deviceSeq +"_"+
                    time);

            mw.put(new IntWritable(3),value);
            context.write(outKey, mw);
        }
    }

    /**
     * 设备综合效率OEE报表-Mapper
     * 4.设备故障
     */
    public static class DeviceComplexEfficiencyOeeMalfunctionMapper extends Mapper<LongWritable,Text,Text, MapWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            MapWritable mw = new MapWritable();
            // -- 构造key
            String[] datas = value.toString().split(",");
            //设备SEQ
            String deviceSeq = datas[0];
            //报修时间（2019-10-01）
            String time = datas[1].substring(0,10);
            //定义key
            Text outKey = new Text(deviceSeq +"_"+
                    time);
            mw.put(new IntWritable(4),value);
            context.write(outKey, mw);

        }
    }
    /**
     * 设备综合效率OEE报表-Reducer
     */
    public static class DeviceComplexEfficiencyOeeReducer extends Reducer<Text, MapWritable, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<MapWritable> values, Context context) throws IOException, InterruptedException {
            //设备编码
            String deviceCode = "";
            //设备名称
            String deviceName = "";
            //设备类型
            String deviceType = "";
            //设备型号
            String deviceModel = "";
            //上班时间 (分钟)
            BigDecimal workTime = new BigDecimal(960);
            //待机时间
            BigDecimal standbyTime = new BigDecimal("0");
            //加工时间
            BigDecimal machiningTime = new BigDecimal("0");
            //故障时间
            BigDecimal malfunctionTime = new BigDecimal("0");
            //总零件数
            int total = 0;
            //不合格的工件数量
            int failedNum = 0;
            //时间
            String time = "";
            DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            for (MapWritable mw : values){
                //设备OEE
                if (mw.containsKey(new IntWritable(1))){
                    String[] datas = mw.get(new IntWritable(1)).toString().split(",");
                    standbyTime = standbyTime.add(new BigDecimal(datas[5]));
                    machiningTime = machiningTime.add(new BigDecimal(datas[6]));
                    deviceCode = datas[2];
                    deviceName = datas[1];
                    deviceType = datas[4];
                    deviceModel = datas[3];
                    time = datas[7].substring(0,10);
                }
                //不合格数量
                if (mw.containsKey(new IntWritable(2))){
                    String[] datas = mw.get(new IntWritable(2)).toString().split(",");
                    deviceCode = datas[1];
                    deviceName = datas[2];
                    deviceType = datas[4];
                    deviceModel = datas[3];
                    time = datas[5].substring(0,10);
                    failedNum++;
                }
                //总零件数
                if (mw.containsKey(new IntWritable(3))){
                    String[] datas = mw.get(new IntWritable(3)).toString().split(",");
                    deviceCode = datas[1];
                    deviceName = datas[2];
                    deviceType = datas[4];
                    deviceModel = datas[3];
                    time = datas[5].substring(0,10);
                    total++;
                }
                //故障时间
                if (mw.containsKey(new IntWritable(4))){
                    String[] datas = mw.get(new IntWritable(4)).toString().split(",");
                    deviceCode = datas[3];
                    deviceName = datas[4];
                    deviceType = datas[6];
                    deviceModel = datas[5];
                    time = datas[1].substring(0,10);
                    //4.故障时间
                    //维修完成时间不为空
                    if (! datas[2].equals("null")) {
                        Date d1 = null;
                        Date d2 = null;
                        try {
                            //维修结束时间
                            d1 = df.parse(datas[2]);
                            //报修时间
                            d2 = df.parse(datas[1]);
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }
                        long diff = 0L;
                        long minutes = 0L;
                        //FIXME 维修完成时间可能小于报修时间，时间差值是负数处理成0
                        if (d1.getTime() > d2.getTime()){
                           diff = d1.getTime() - d2.getTime();//这样得到的差值是毫秒级别
                            minutes = (diff/(1000* 60));
                        }
                        malfunctionTime = malfunctionTime.add(new BigDecimal(minutes));
                    }
                }
            }

            //统计后的数据是秒，需要换算成分钟
            standbyTime = standbyTime.divide(new BigDecimal("60"),0,BigDecimal.ROUND_HALF_UP);
            machiningTime = machiningTime.divide(new BigDecimal("60"),0,BigDecimal.ROUND_HALF_UP);

            Text outValue = new Text(deviceCode + ","+
                    deviceName + ","+
                    deviceModel + ","+
                    deviceType + ","+
                    standbyTime + ","+
                    workTime + ","+
                    machiningTime + ","+
                    malfunctionTime + ","+
                    total+ ","+
                    failedNum+ ","+
                    time);
            Text outKey = new Text(DigestUtils.shaHex(key.toString()));
            context.write(outKey, outValue);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("mapred.textoutputformat.separator", ",");

        // 如果输出目录已经存在，则先删除
        FileSystem fileSystem = FileSystem.get(configuration);

        // 输出文件
        Path outputPath = new Path(args[0]);
        if (fileSystem.exists(outputPath)) {
            fileSystem.delete(outputPath, true);
        }
        // 输入文件
        Path inputPath1 = new Path(args[1]);
        Path inputPath2 = new Path(args[2]);
        Path inputPath3 = new Path(args[3]);
        Path inputPath4 = new Path(args[4]);
        // job
        Job job = Job.getInstance(configuration);
        job.setJobName("DeviceComplexEfficiencyOeeJob");
        job.setJarByClass(DeviceComplexEfficiencyOeeJob.class);

        // 多个输入对应多个mapper
        //OEE
        MultipleInputs.addInputPath(job,inputPath1, TextInputFormat.class, DeviceComplexEfficiencyOeeMapper.class);
        //不合格的零件数量
        MultipleInputs.addInputPath(job,inputPath2, TextInputFormat.class, DeviceComplexEfficiencyOeeUnqualifiedMapper.class);
        //总零件数
        MultipleInputs.addInputPath(job,inputPath3, TextInputFormat.class, DeviceComplexEfficiencyOeeTotalMapper.class);
        //设备故障
        MultipleInputs.addInputPath(job,inputPath4, TextInputFormat.class, DeviceComplexEfficiencyOeeMalfunctionMapper.class);

        //map的outKey outValue类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(MapWritable.class);
        // reducer的outKey  outValue类型
        job.setReducerClass(DeviceComplexEfficiencyOeeReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(MapWritable.class);
        // 一个输出
        FileOutputFormat.setOutputPath(job,outputPath);
        // 完成
        job.waitForCompletion(true);
        System.out.println("---设备综合效率OEE报表Job完成---");
    }
}
